package com.health.trans.channel;


import cn.hutool.core.exceptions.ExceptionUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.util.Collection;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 内存通道
 * 内存Channel的具体实现，底层其实是一个ArrayBlockingQueue
 *
 * @author nqlz
 * @date 2022/04/24
 */
@Slf4j
@Data
public class Channel<T> {
    protected volatile long waitReaderTime = 0;

    protected volatile long waitWriterTime = 0;
    private int bufferSize = 0;
    protected int capacity = 2048;

    private ArrayBlockingQueue<T> queue = null;

    private ReentrantLock lock;

    private Condition notInsufficient, notEmpty;

    public Channel(int bufferSize,int capacity) {
        this.queue = new ArrayBlockingQueue<>(capacity);
        this.bufferSize = bufferSize;
        lock = new ReentrantLock();
        notInsufficient = lock.newCondition();
        notEmpty = lock.newCondition();
    }

    public void clear() {
        this.queue.clear();
    }

    protected void push(T r) {
        try {
            long startTime = System.nanoTime();
            this.queue.put(r);
            waitWriterTime += System.nanoTime() - startTime;
        } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
        }
    }

    protected void pushAll(Collection<T> rs) {
        try {
            long startTime = System.nanoTime();
            lock.lockInterruptibly();
            while (rs.size() > this.queue.remainingCapacity()) {
                notInsufficient.await(200L, TimeUnit.MILLISECONDS);
            }
            this.queue.addAll(rs);
            waitWriterTime += System.nanoTime() - startTime;
            notEmpty.signalAll();
        } catch (InterruptedException e) {
            ExceptionUtil.wrapAndThrow(e);
        } finally {
            lock.unlock();
        }
    }

    protected T pull() {
        try {
            long startTime = System.nanoTime();
            T r = this.queue.take();
            waitReaderTime += System.nanoTime() - startTime;
            return r;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(e);
        }
    }

    protected void pullAll(Collection<T> rs) {
        assert rs != null;
        rs.clear();
        try {
            long startTime = System.nanoTime();
            lock.lockInterruptibly();
            while (this.queue.drainTo(rs, bufferSize) <= 0) {
                notEmpty.await(200L, TimeUnit.MILLISECONDS);
            }
            waitReaderTime += System.nanoTime() - startTime;
            notInsufficient.signalAll();
        } catch (InterruptedException e) {
            ExceptionUtil.wrapAndThrow(e);
        } finally {
            lock.unlock();
        }
    }

    public int size() {
        return this.queue.size();
    }

    public boolean isEmpty() {
        return this.queue.isEmpty();
    }
}
